#!usr/bin/env python  
# -*- coding:utf-8 _*-
""" 
@author:robot
@file: ct_task.py 
@version:
@time: 2022/03/11 
"""
import random
import time

import allure

from TapException.CONNECTOR import ConnectorInitFaild
from TapException.PARALLEL import *
from TapClientCore.T01_init_conf.v2_init.init_config import timeout_config, MFA, parallel_record_collection
from TapClientConnector import TapClientConnector
from TapClientCore.T06_Tapdata_feature.v2.TPF03_task.tpf_task_v1 import V2TPFTask, logger
from TapClientCore.T06_Tapdata_feature.v2.TPF04_inspects.tpf_inspects_v1 import V2TPFInspects
from TapClientCore.T06_Tapdata_feature.v2.TPF05_meta.tpf_meta_node_schema import V2TPFMetaNodeSchema
from TapException.TM import *

check_wait = 5
cdc_wait = 180


# 并行执行
class CTTaskUtilParallel(V2TPFTask, V2TPFInspects, V2TPFMetaNodeSchema):
    @staticmethod
    def random_wait():
        _ = random.randint(1, check_wait * 3)
        time.sleep(_)

    @staticmethod
    def lock_and_only_get_table(data):
        prefix = data.get('prefix')
        suffix = data.get('suffix')
        if not prefix:
            prefix = ''
        if not suffix:
            suffix = ''
        task_type = data.get('task_type')
        _table = ''
        # 获取数据源配置信息的name
        db_name = data.get('operate').get('source').get('source_info').get('name')
        parallel_use_record_name = data.get('parallel_use_record_name')
        if parallel_use_record_name:
            parallel_record = MFA.query(
                parallel_record_collection,
                {
                    'name': parallel_use_record_name,
                    'is_use': '0',
                    'db_name': db_name
                }
            )
            if parallel_record:
                _record = parallel_record[0]
                _id = _record.get('_id')
                _name = _record.get('name')
                _table = _record.get('table')
                _table_list = _table
                if task_type == 'data copy':
                    if not isinstance(_table, list):
                        _table_list = [_table]
                    data['task_config']['relationship']['syncObjects']['objectNames'] = _table_list
                    data['operate']['source']['table'] = _table
                    data['operate']['target']['table'] = _table
                elif 'data development' in task_type:
                    target_table = f'{prefix}{_table}{suffix}'
                    data['task_config']['relationship']['source'].update({'name': _table, 'tableName': _table})
                    data['task_config']['relationship']['sink'].update(
                        {'name': target_table, 'tableName': target_table})
                    data['operate']['source']['table'] = _table
                    data['operate']['target']['table'] = target_table
                return data
            else:
                raise NotCanUseRecord
        else:
            logger.info('not parallel execute')
            return data

    @staticmethod
    def lock_and_get_table(data):
        prefix = data.get('prefix')
        suffix = data.get('suffix')
        if not prefix:
            prefix = ''
        if not suffix:
            suffix = ''
        task_type = data.get('task_type')
        _table = ''
        # 获取数据源配置信息的name
        db_name = data.get('operate').get('source').get('source_info').get('name')
        parallel_use_record_name = data.get('parallel_use_record_name')
        if parallel_use_record_name:
            parallel_record = MFA.query(
                parallel_record_collection,
                {
                    'name': parallel_use_record_name,
                    'is_use': '0',
                    'db_name': db_name
                }
            )
            if parallel_record:
                _record = parallel_record[0]
                _id = _record.get('_id')
                _name = _record.get('name')
                _table = _record.get('table')
                _table_list = _table
                if task_type == 'data copy':
                    if not isinstance(_table, list):
                        _table_list = [_table]
                    data['task_config']['relationship']['syncObjects']['objectNames'] = _table_list
                    data['operate']['source']['table'] = _table
                    data['operate']['target']['table'] = _table
                elif 'data development' in task_type:
                    target_table = f'{prefix}{_table}{suffix}'
                    data['task_config']['relationship']['source'].update({'name': _table, 'tableName': _table})
                    data['task_config']['relationship']['sink'].update(
                        {'name': target_table, 'tableName': target_table})
                    data['operate']['source']['table'] = _table
                    data['operate']['target']['table'] = target_table
                MFA.update(
                    parallel_record_collection,
                    {'_id': _id},
                    {'$set': {'is_use': '1'}}
                )
                data['id'] = _id
                return data
            else:
                raise NotCanUseRecord
        else:
            logger.info('not parallel execute')
            return data

    @staticmethod
    def dummy_get_table(data):
        prefix = data.get('prefix')
        suffix = data.get('suffix')
        if not prefix:
            prefix = ''
        if not suffix:
            suffix = ''
        task_type = data.get('task_type')
        _table = ''
        # 获取数据源配置信息的name
        if task_type == 'data copy':
            _table = data['operate']['source']['table']
            _table_list = _table
            if not isinstance(_table, list):
                _table_list = [_table]
            data['task_config']['relationship']['syncObjects']['objectNames'] = _table_list
            data['operate']['target']['table'] = _table
        elif 'data development' in task_type:
            _table = data['operate']['source']['table']
            target_table = f'{prefix}{_table}{suffix}'
            data['task_config']['relationship']['source'].update({'name': _table, 'tableName': _table})
            data['task_config']['relationship']['sink'].update(
                {'name': target_table, 'tableName': target_table})
            data['operate']['target']['table'] = target_table
        return data

    @staticmethod
    def release_table(_id):
        with allure.step('释放使用的表'):
            if _id:
                MFA.update(
                    parallel_record_collection,
                    {'_id': _id},
                    {'$set': {'is_use': '0'}}
                )
                logger.info(f'release {_id} record')


# connector
class CTTaskUtilConnector(CTTaskUtilParallel):
    @staticmethod
    def init_connector(db_info):
        return TapClientConnector(db_info)

    @staticmethod
    def tap_query(connector: [TapClientConnector], **args):
        order = 'order by "pk"'
        if connector.database_type in ['mysql', 'mariadb', 'mysql pxc',
                                       'tidb', 'kundb', 'adb_mysql',
                                       'doris', 'oceanbase', 'gbase-8a']:
            order = 'order by pk'
        dml = connector.tapclient_connector.generate_select(**args, order=order)
        return connector.tapclient_connector.query(dml)

    @staticmethod
    def tap_count_query(connector: [TapClientConnector], **args):
        dml = connector.tapclient_connector.generate_select(**args)
        return connector.tapclient_connector.query(dml)

    @staticmethod
    def tap_delete(connector: [TapClientConnector], **args):
        dml = connector.tapclient_connector.generate_delete(**args)
        return connector.tapclient_connector.exec(dml)

    @staticmethod
    def tap_insert(connector: [TapClientConnector], **args):
        dml = connector.tapclient_connector.generate_insert(**args)
        return connector.tapclient_connector.exec(dml)

    @staticmethod
    def tap_update(connector: [TapClientConnector], **args):
        dml = connector.tapclient_connector.generate_update(**args)
        return connector.tapclient_connector.exec(dml)


# check
class CTTaskUtilCheck(CTTaskUtilConnector):
    def count_check(self, source_info, source_table, target_info, sink_table):
        time.sleep(check_wait)
        with allure.step('count校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            source_result = self.tap_count_query(source_connector, table=source_table, column='count(*)')
            sink_result = self.tap_count_query(sink_connector, table=sink_table, column='count(*)')
        logger.info(f'source count -> {source_result}')
        logger.info(f'sink count -> {sink_result}')
        assert source_result == sink_result

    def cdc_count_check(self, source_info, source_table, source_where, target_info, sink_table):
        time.sleep(check_wait)
        with allure.step('count校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            source_result = self.tap_count_query(source_connector, table=source_table, column='count(*)',
                                                 where=source_where)
            sink_result = self.tap_count_query(sink_connector, table=sink_table, column='count(*)')
        logger.info(f'source count -> {source_result}')
        logger.info(f'sink count -> {sink_result}')
        assert source_result == sink_result

    def tap_check(self, source_info, source_table, target_info, sink_table):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                assert source_result == sink_result
            elif sink_connector.database_type == 'redis' and sink_connector.tapclient_connector.data_struct == 'json':
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in sink_result:
                    tmp_2 = []
                    for i in source_connector.tapclient_connector.TABLE_DESCRIPTION:
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(source_result)
                logger.info(tmp_1)
                assert source_result == tmp_1
            elif source_connector.database_type in ['kafka', 'activemq', 'rabbitmq', 'rocketmq']:
                tmp_1 = []
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                if sink_connector.database_type in ['mongodb']:
                    for j in sink_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(source_result)
                    logger.info(tmp_1)
                    assert source_result == tmp_1
                elif sink_connector.database_type in ['es']:
                    logger.info(source_result)
                    logger.info(sink_result)
                    assert source_result == sink_result
                else:
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
            elif source_connector.database_type in ['mongodb']:
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                if sink_connector.database_type in ['kafka', 'activemq', 'rabbitmq', 'rocketmq']:
                    source_result = self.tap_query(source_connector, table=source_table, column=None)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
                elif sink_connector.database_type in ['elasticsearch']:
                    source_result = self.tap_query(source_connector, table=source_table)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
                else:
                    source_result = self.tap_query(source_connector, table=source_table, column=None)
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
            elif sink_connector.database_type in [
                'mongodb', 'elasticsearch', 'kafka',
                'activemq', 'rabbitmq', 'rocketmq'
            ]:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in sink_result:
                    tmp_2 = []
                    for i in source_connector.tapclient_connector.TABLE_DESCRIPTION:
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(source_result)
                logger.info(tmp_1)
                assert source_result == tmp_1
            else:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                assert source_result == sink_result

    def tap_check_mq_kafka(self, source_data, target_info, sink_table):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            sink_connector = self.init_connector(target_info)
            if sink_connector.database_type in ['mongodb', 'elasticsearch', 'kafka',
                                                'activemq', 'rabbitmq', 'rocketmq']:
                source_result = source_data
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                assert source_result == sink_result
            else:
                source_result = source_data
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in source_result:
                    tmp_2 = []
                    for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(tmp_1)
                logger.info(sink_result)
                assert tmp_1 == sink_result

    def tap_cdc_check(self, source_info, source_table, source_where, target_info, sink_table):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                assert source_result == sink_result
            elif source_connector.database_type in ['kafka']:
                tmp_1 = []
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                if sink_connector.database_type in ['mongodb']:
                    for j in sink_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(source_result)
                    logger.info(tmp_1)
                    assert source_result == tmp_1
                elif sink_connector.database_type in ['es']:
                    logger.info(source_result)
                    logger.info(sink_result)
                    assert source_result == sink_result
                else:
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
            elif source_connector.database_type in ['mongodb']:
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                if sink_connector.database_type in ['kafka', 'activemq', 'rabbitmq', 'rocketmq']:
                    source_result = self.tap_query(source_connector, table=source_table,
                                                   where=source_where, column=None)

                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
                elif sink_connector.database_type in ['elasticsearch']:
                    source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
                else:
                    source_result = self.tap_query(source_connector, table=source_table, where=source_where,
                                                   column=None)
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    assert tmp_1 == sink_result
            elif sink_connector.database_type in [
                'mongodb', 'elasticsearch', 'kafka',
                'activemq', 'rabbitmq', 'rocketmq'
            ]:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in sink_result:
                    tmp_2 = []
                    for i in source_connector.tapclient_connector.TABLE_DESCRIPTION:
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(source_result)
                logger.info(tmp_1)
                assert source_result == tmp_1
            else:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                assert source_result == sink_result

    def tap_more_table_check(self, source_info, source_table_list, target_info):
        time.sleep(check_wait)
        for source_table in source_table_list:
            sink_table = source_table
            with allure.step('数据校验'):
                source_connector = self.init_connector(source_info)
                sink_connector = self.init_connector(target_info)
                if source_connector.database_type == sink_connector.database_type:
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    logger.info(source_result)
                    logger.info(sink_result)
                    assert source_result == sink_result
                elif source_connector.database_type in ['kafka']:
                    tmp_1 = []
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    if sink_connector.database_type in ['mongodb']:
                        for j in sink_result:
                            tmp_2 = {}
                            for k in j:
                                from bson import Decimal128
                                from bson import ObjectId
                                if isinstance(j[k], Decimal128):
                                    tmp_2[k] = int(j[k].to_decimal())
                                if isinstance(j[k], ObjectId):
                                    tmp_2[k] = str(j[k])
                                else:
                                    tmp_2[k] = j[k]
                            tmp_1.append(tmp_2)
                        logger.info(source_result)
                        logger.info(tmp_1)
                        assert source_result == tmp_1
                    elif sink_connector.database_type in ['es']:
                        logger.info(source_result)
                        logger.info(sink_result)
                        assert source_result == sink_result
                    else:
                        for j in source_result:
                            tmp_2 = []
                            for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                                for k in j:
                                    if i == k:
                                        t = j[k]
                                        tmp_2.append(t)
                            tmp_1.append(tuple(tmp_2))
                        logger.info(tmp_1)
                        logger.info(sink_result)
                        assert tmp_1 == sink_result
                elif source_connector.database_type in ['mongodb']:
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    tmp_1 = []
                    if sink_connector.database_type in ['kafka', 'activemq', 'rabbitmq', 'rocketmq']:
                        source_result = self.tap_query(source_connector, table=source_table, column=None)
                        for j in source_result:
                            tmp_2 = {}
                            for k in j:
                                from bson import Decimal128
                                from bson import ObjectId
                                if isinstance(j[k], Decimal128):
                                    tmp_2[k] = int(j[k].to_decimal())
                                if isinstance(j[k], ObjectId):
                                    tmp_2[k] = str(j[k])
                                else:
                                    tmp_2[k] = j[k]
                            tmp_1.append(tmp_2)
                        logger.info(tmp_1)
                        logger.info(sink_result)
                        assert tmp_1 == sink_result
                    elif sink_connector.database_type in ['elasticsearch']:
                        source_result = self.tap_query(source_connector, table=source_table)
                        for j in source_result:
                            tmp_2 = {}
                            for k in j:
                                from bson import Decimal128
                                from bson import ObjectId
                                if isinstance(j[k], Decimal128):
                                    tmp_2[k] = int(j[k].to_decimal())
                                if isinstance(j[k], ObjectId):
                                    tmp_2[k] = str(j[k])
                                else:
                                    tmp_2[k] = j[k]
                            tmp_1.append(tmp_2)
                        logger.info(tmp_1)
                        logger.info(sink_result)
                        assert tmp_1 == sink_result
                    else:
                        source_result = self.tap_query(source_connector, table=source_table, column=None)
                        for j in source_result:
                            tmp_2 = []
                            for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                                for k in j:
                                    if i == k:
                                        t = j[k]
                                        from bson import Decimal128
                                        if isinstance(j[k], Decimal128):
                                            t = int(j[k].to_decimal())
                                        from bson import ObjectId
                                        if isinstance(j[k], ObjectId):
                                            t = str(j[k])
                                        tmp_2.append(t)
                            tmp_1.append(tuple(tmp_2))
                        logger.info(tmp_1)
                        logger.info(sink_result)
                        assert tmp_1 == sink_result
                elif sink_connector.database_type in [
                    'mongodb', 'elasticsearch', 'kafka',
                    'activemq', 'rabbitmq', 'rocketmq'
                ]:
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    tmp_1 = []
                    for j in sink_result:
                        tmp_2 = []
                        for i in source_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(source_result)
                    logger.info(tmp_1)
                    assert source_result == tmp_1
                else:
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    logger.info(source_result)
                    logger.info(sink_result)
                    assert source_result == sink_result

    async def async_tap_more_table_check(self, source_info, source_table_list, target_info):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            for source_table in source_table_list:
                sink_table = source_table
                source_connector = self.init_connector(source_info)
                sink_connector = self.init_connector(target_info)
                if source_connector.database_type == sink_connector.database_type:
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    logger.info(source_result)
                    logger.info(sink_result)
                    assert source_result == sink_result
                elif source_connector.database_type in ['kafka']:
                    tmp_1 = []
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    if sink_connector.database_type in ['mongodb']:
                        for j in sink_result:
                            tmp_2 = {}
                            for k in j:
                                from bson import Decimal128
                                from bson import ObjectId
                                if isinstance(j[k], Decimal128):
                                    tmp_2[k] = int(j[k].to_decimal())
                                if isinstance(j[k], ObjectId):
                                    tmp_2[k] = str(j[k])
                                else:
                                    tmp_2[k] = j[k]
                            tmp_1.append(tmp_2)
                        logger.info(source_result)
                        logger.info(tmp_1)
                        assert source_result == tmp_1
                    elif sink_connector.database_type in ['es']:
                        logger.info(source_result)
                        logger.info(sink_result)
                        assert source_result == sink_result
                    else:
                        for j in source_result:
                            tmp_2 = []
                            for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                                for k in j:
                                    if i == k:
                                        t = j[k]
                                        tmp_2.append(t)
                            tmp_1.append(tuple(tmp_2))
                        logger.info(tmp_1)
                        logger.info(sink_result)
                        assert tmp_1 == sink_result
                elif source_connector.database_type in ['mongodb']:
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    tmp_1 = []
                    if sink_connector.database_type in ['kafka', 'activemq', 'rabbitmq', 'rocketmq']:
                        source_result = self.tap_query(source_connector, table=source_table, column=None)
                        for j in source_result:
                            tmp_2 = {}
                            for k in j:
                                from bson import Decimal128
                                from bson import ObjectId
                                if isinstance(j[k], Decimal128):
                                    tmp_2[k] = int(j[k].to_decimal())
                                if isinstance(j[k], ObjectId):
                                    tmp_2[k] = str(j[k])
                                else:
                                    tmp_2[k] = j[k]
                            tmp_1.append(tmp_2)
                        logger.info(tmp_1)
                        logger.info(sink_result)
                        assert tmp_1 == sink_result
                    elif sink_connector.database_type in ['elasticsearch']:
                        source_result = self.tap_query(source_connector, table=source_table)
                        for j in source_result:
                            tmp_2 = {}
                            for k in j:
                                from bson import Decimal128
                                from bson import ObjectId
                                if isinstance(j[k], Decimal128):
                                    tmp_2[k] = int(j[k].to_decimal())
                                if isinstance(j[k], ObjectId):
                                    tmp_2[k] = str(j[k])
                                else:
                                    tmp_2[k] = j[k]
                            tmp_1.append(tmp_2)
                        logger.info(tmp_1)
                        logger.info(sink_result)
                        assert tmp_1 == sink_result
                    else:
                        source_result = self.tap_query(source_connector, table=source_table, column=None)
                        for j in source_result:
                            tmp_2 = []
                            for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                                for k in j:
                                    if i == k:
                                        t = j[k]
                                        from bson import Decimal128
                                        if isinstance(j[k], Decimal128):
                                            t = int(j[k].to_decimal())
                                        from bson import ObjectId
                                        if isinstance(j[k], ObjectId):
                                            t = str(j[k])
                                        tmp_2.append(t)
                            tmp_1.append(tuple(tmp_2))
                        logger.info(tmp_1.sort())
                        logger.info(sink_result.sort())
                        assert tmp_1 == sink_result
                elif sink_connector.database_type in [
                    'mongodb', 'elasticsearch', 'kafka',
                    'activemq', 'rabbitmq', 'rocketmq'
                ]:
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    tmp_1 = []
                    for j in sink_result:
                        tmp_2 = []
                        for i in source_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(source_result)
                    logger.info(tmp_1)
                    assert source_result == tmp_1
                else:
                    source_result = self.tap_query(source_connector, table=source_table)
                    sink_result = self.tap_query(sink_connector, table=sink_table)
                    logger.info(source_result)
                    logger.info(sink_result)
                    assert source_result == sink_result

    @staticmethod
    def drop_target_table(connector: TapClientConnector):
        logger.info('开始进入目标表删除阶段')
        table = connector.tapclient_connector.CURRENT_TABLE
        if connector.database_type in ['elasticsearch', 'gbase-8s', 'gbase-8a']:
            table = table.lower()
        elif connector.database_type in ['activemq', 'rabbitmq', 'rocketmq']:
            # # 查询数据情况数据
            if connector.database_type in ['rabbitmq']:
                connector.tapclient_connector.timeout = 5
                connector.tapclient_connector.drop(table)
            return
        elif connector.database_type in ['redis']:
            if connector.tapclient_connector.data_struct == 'list':
                for raw_table in connector.tapclient_connector.show_tables():
                    if table == raw_table:
                        with allure.step('目标表删除'):
                            logger.info(f'发现目标要删除的表{table}')
                            drop_operate = connector.tapclient_connector.generate_drop(table)
                            connector.tapclient_connector.exec(drop_operate)
                            logger.info(f'删除目标要同步的表{table}')
                            return
            elif connector.tapclient_connector.data_struct == 'json':
                for raw_table in connector.tapclient_connector.show_tables():
                    if table in raw_table:
                        with allure.step('目标表删除'):
                            logger.info(f'发现目标要删除的表{raw_table}')
                            drop_operate = connector.tapclient_connector.generate_drop(raw_table)
                            connector.tapclient_connector.exec(drop_operate)
                            logger.info(f'删除目标要同步的表{raw_table}')
                return
        for raw_table in connector.tapclient_connector.show_tables():
            if table == raw_table:
                with allure.step('目标表删除'):
                    logger.info(f'发现目标要删除的表{table}')
                    drop_operate = connector.tapclient_connector.generate_drop(table)
                    connector.tapclient_connector.exec(drop_operate)
                    logger.info(f'删除目标要同步的表{table}')
                    return
        logger.info(f'没有发现目标要删除的表{table}')

    @staticmethod
    def drop_more_target_table(connector: TapClientConnector, table_list):
        logger.info('开始进入目标表删除阶段')
        for table in table_list:
            if connector.database_type in ['elasticsearch', 'gbase-8s', 'gbase-8a']:
                table = table.lower()
            for raw_table in connector.tapclient_connector.show_tables():
                if table == raw_table:
                    with allure.step('目标表删除'):
                        logger.info(f'发现目标要删除的表{table}')
                        drop_operate = connector.tapclient_connector.generate_drop(table)
                        connector.tapclient_connector.exec(drop_operate)
                        logger.info(f'删除目标要同步的表{table}')
                        break
            logger.info(f'没有发现目标要删除的表{table}')

    @staticmethod
    async def async_drop_table(connector: TapClientConnector, table_list):
        try:
            for table in table_list:
                if connector.database_type in ['elasticsearch', 'gbase-8s', 'gbase-8a']:
                    table = table.lower()
                drop_sql = connector.tapclient_connector.generate_drop(table)
                connector.tapclient_connector.exec(drop_sql)
                logger.info('drop success')
        except Exception as e:
            logger.warning('drop exception')


# node check
class CTTaskUtilNodeCheck(CTTaskUtilCheck):
    # for dev operate node
    @staticmethod
    def tap_query_get_column(connector, **args):
        dml = connector.tapclient_connector.generate_select(**args, order='order by "pk"')
        return connector.tapclient_connector.query_get_column(dml)

    def column_check(self, source_info, source_table, target_info, sink_table, source_func):
        time.sleep(check_wait)
        with allure.step('rename fieldname check'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type in ['mongodb']:
                source_result = self.tap_query_get_column(source_connector, table=source_table, column=None)
                if sink_connector.database_type in ['mongodb']:
                    sink_result = self.tap_query_get_column(sink_connector, table=sink_table, column=None)
                else:
                    sink_result = self.tap_query_get_column(sink_connector, table=sink_table)
            else:
                source_result = self.tap_query_get_column(source_connector, table=source_table)
                sink_result = self.tap_query_get_column(sink_connector, table=sink_table)
            logger.info(source_result)
            logger.info('column rename')
            source_transform = source_func(source_result)
            source_transform.sort()
            sink_result.sort()
            logger.info(source_transform)
            logger.info(sink_result)
            assert source_transform == sink_result
            return source_transform

    def column_agg_check(self, source_info, source_table, target_info, sink_table, source_func):
        time.sleep(check_wait)
        with allure.step('rename fieldname check'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            source_result = self.tap_query_get_column(source_connector, table=source_table)
            sink_result = self.tap_query_get_column(sink_connector, table=sink_table)
            logger.info(source_result)
            logger.info('column rename')
            source_transform = source_func(source_result)
            source_transform.sort()
            sink_result.sort()
            logger.info(source_transform)
            logger.info(sink_result)
            assert source_transform == sink_result
            return source_transform

    def tap_node_check(self, source_info, source_table, target_info, sink_table, func):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result
            elif source_connector.database_type in ['kafka']:
                tmp_1 = []
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                if sink_connector.database_type in ['mongodb']:
                    for j in sink_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(source_result)
                    logger.info(tmp_1)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == tmp_1
                elif sink_connector.database_type in ['es']:
                    logger.info(source_result)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == sink_result
                else:
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif source_connector.database_type in ['mongodb']:
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                if sink_connector.database_type in ['kafka']:
                    source_result = self.tap_query(source_connector, table=source_table, column=None)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                elif sink_connector.database_type in ['elasticsearch']:
                    source_result = self.tap_query(source_connector, table=source_table)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                else:
                    source_result = self.tap_query(source_connector, table=source_table, column=None)
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif sink_connector.database_type in ['mongodb', 'elasticsearch', 'kafka']:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in sink_result:
                    tmp_2 = []
                    for i in source_connector.tapclient_connector.TABLE_DESCRIPTION:
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(source_result)
                logger.info(tmp_1)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == tmp_1
            else:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result

    def tap_join_node_check(self, source_info, source_table, target_info, sink_table, func, column_func):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result
            elif source_connector.database_type in ['kafka']:
                tmp_1 = []
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                if sink_connector.database_type in ['mongodb']:
                    for j in sink_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(source_result)
                    logger.info(tmp_1)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == tmp_1
                elif sink_connector.database_type in ['es']:
                    logger.info(source_result)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == sink_result
                else:
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif source_connector.database_type in ['mongodb']:
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                if sink_connector.database_type in ['kafka']:
                    source_result = self.tap_query(source_connector, table=source_table, column=None)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                elif sink_connector.database_type in ['elasticsearch']:
                    source_result = self.tap_query(source_connector, table=source_table)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                else:
                    source_result = self.tap_query(source_connector, table=source_table, column=None)
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif sink_connector.database_type in ['mongodb', 'elasticsearch', 'kafka']:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in sink_result:
                    tmp_2 = []
                    for i in column_func(source_connector.tapclient_connector.TABLE_DESCRIPTION):
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(source_result)
                logger.info(tmp_1)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == tmp_1
            else:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result

    def tap_node_cdc_check(self, source_info, source_table, source_where, target_info, sink_table, func):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result
            elif source_connector.database_type in ['kafka']:
                tmp_1 = []
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                if sink_connector.database_type in ['mongodb']:
                    for j in sink_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(source_result)
                    logger.info(tmp_1)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == tmp_1
                elif sink_connector.database_type in ['es']:
                    logger.info(source_result)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == sink_result
                else:
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif source_connector.database_type in ['mongodb']:
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                if sink_connector.database_type in ['kafka']:
                    source_result = self.tap_query(source_connector, table=source_table, column=None,
                                                   where=source_where)

                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                elif sink_connector.database_type in ['elasticsearch']:
                    source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                else:
                    source_result = self.tap_query(source_connector, table=source_table, where=source_where,
                                                   column=None)
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif sink_connector.database_type in ['mongodb', 'elasticsearch', 'kafka']:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in sink_result:
                    tmp_2 = []
                    for i in source_connector.tapclient_connector.TABLE_DESCRIPTION:
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(source_result)
                logger.info(tmp_1)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == tmp_1
            else:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result

    def tap_join_node_cdc_check(self, source_info, source_table, source_where, target_info, sink_table, func,
                                column_func):
        time.sleep(check_wait)
        with allure.step('数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result
            elif source_connector.database_type in ['kafka']:
                tmp_1 = []
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                if sink_connector.database_type in ['mongodb']:
                    for j in sink_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(source_result)
                    logger.info(tmp_1)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == tmp_1
                elif sink_connector.database_type in ['es']:
                    logger.info(source_result)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(source_result))
                    assert func(source_result) == sink_result
                else:
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif source_connector.database_type in ['mongodb']:
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                if sink_connector.database_type in ['kafka']:
                    source_result = self.tap_query(source_connector, table=source_table, column=None,
                                                   where=source_where)

                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                elif sink_connector.database_type in ['elasticsearch']:
                    source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                    for j in source_result:
                        tmp_2 = {}
                        for k in j:
                            from bson import Decimal128
                            from bson import ObjectId
                            if isinstance(j[k], Decimal128):
                                tmp_2[k] = int(j[k].to_decimal())
                            if isinstance(j[k], ObjectId):
                                tmp_2[k] = str(j[k])
                            else:
                                tmp_2[k] = j[k]
                        tmp_1.append(tmp_2)
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
                else:
                    source_result = self.tap_query(source_connector, table=source_table, where=source_where,
                                                   column=None)
                    for j in source_result:
                        tmp_2 = []
                        for i in sink_connector.tapclient_connector.TABLE_DESCRIPTION:
                            for k in j:
                                if i == k:
                                    t = j[k]
                                    from bson import Decimal128
                                    if isinstance(j[k], Decimal128):
                                        t = int(j[k].to_decimal())
                                    from bson import ObjectId
                                    if isinstance(j[k], ObjectId):
                                        t = str(j[k])
                                    tmp_2.append(t)
                        tmp_1.append(tuple(tmp_2))
                    logger.info(tmp_1)
                    logger.info(sink_result)
                    logger.info('func convert')
                    logger.info(func(tmp_1))
                    assert func(tmp_1) == sink_result
            elif sink_connector.database_type in ['mongodb', 'elasticsearch', 'kafka']:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                tmp_1 = []
                for j in sink_result:
                    tmp_2 = []
                    for i in column_func(source_connector.tapclient_connector.TABLE_DESCRIPTION):
                        for k in j:
                            if i == k:
                                t = j[k]
                                from bson import Decimal128
                                if isinstance(j[k], Decimal128):
                                    t = int(j[k].to_decimal())
                                from bson import ObjectId
                                if isinstance(j[k], ObjectId):
                                    t = str(j[k])
                                tmp_2.append(t)
                    tmp_1.append(tuple(tmp_2))
                logger.info(source_result)
                logger.info(tmp_1)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == tmp_1
            else:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result))
                assert func(source_result) == sink_result

    def tap_merge_check(
            self,
            source_info,
            source_table,
            target_info,
            sink_table,
            func,
            func2,
            column
    ):
        time.sleep(check_wait)
        with allure.step('merge-f数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table, column=None)
                sink_result = self.tap_query(sink_connector, table=sink_table, column=None)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func2(func(source_result), column))
                logger.info(func2(sink_result, column))
                assert func2(func(source_result), column) == func2(func(source_result), column)
            else:
                source_result = self.tap_query(source_connector, table=source_table)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                source_column = source_connector.tapclient_connector.TABLE_DESCRIPTION
                source_result2 = []
                for one_record in source_result:
                    source_result2.append({
                        # column 与 values 合并为字典
                        source_column[i]: one_record[i] for i in range(len(source_column))
                    })
                logger.info(source_result2)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result2))
                assert func(source_result2) == sink_result

    def tap_merge_cdc_check(
            self,
            source_info,
            source_table,
            source_where,
            target_info,
            sink_table,
            func,
            func2,
            column
    ):
        time.sleep(check_wait)
        with allure.step('merge-f数据校验'):
            source_connector = self.init_connector(source_info)
            sink_connector = self.init_connector(target_info)
            if source_connector.database_type == sink_connector.database_type:
                source_result = self.tap_query(source_connector, table=source_table, column=None, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table, column=None)
                logger.info(source_result)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func2(func(source_result), column))
                logger.info(func2(sink_result, column))
                assert func2(func(source_result), column) == func2(func(source_result), column)
            else:
                source_result = self.tap_query(source_connector, table=source_table, where=source_where)
                sink_result = self.tap_query(sink_connector, table=sink_table)
                source_column = source_connector.tapclient_connector.TABLE_DESCRIPTION
                source_result2 = []
                for one_record in source_result:
                    source_result2.append({
                        # column 与 values 合并为字典
                        source_column[i]: one_record[i] for i in range(len(source_column))
                    })
                logger.info(source_result2)
                logger.info(sink_result)
                logger.info('func convert')
                logger.info(func(source_result2))
                assert func(source_result2) == sink_result


# task status and stats
class CTTaskUtilStatusAndStats(CTTaskUtilNodeCheck):

    @staticmethod
    def func_wait(func1, func2, args, compare, timeout=timeout_config):
        time.sleep(check_wait)
        with allure.step(f'{func1.__name__} 事件统计'):
            for i in range(timeout):
                time.sleep(5)
                try:
                    # stats
                    result1 = func1(args)
                except Exception:
                    raise TaskStatisticsError
                logger.info(f'{func1.__name__} RESULT is {result1} ')
                result2 = func2(args)
                logger.info(f'task status  is {result2} ')
                if result2 == 'error':
                    raise TaskError
                if result1 > compare:
                    raise TaskStatisticsError
                if result1 == compare:
                    assert True
                    return
            raise TimeoutError

    @staticmethod
    def probably_func_wait(func1, func2, args, compare, timeout=timeout_config):
        with allure.step(f'{func1.__name__} 事件统计'):
            for i in range(timeout):
                time.sleep(5)
                try:
                    # stats
                    result1 = func1(args)
                except Exception:
                    raise TaskStatisticsError
                logger.info(f'{func1.__name__} RESULT is {result1} ')
                result2 = func2(args)
                logger.info(f'task status  is {result2} ')
                if result2 == 'error':
                    raise TaskError
                if result1 >= compare:
                    return
            raise TimeoutError

    @staticmethod
    def func_list_wait(func_list, func, args, compare, timeout=timeout_config):
        time.sleep(check_wait)
        with allure.step('事件统计'):
            for i in range(timeout):
                time.sleep(5)
                insert_update = func_list(args)
                result2 = func(args)
                logger.info(f'task status  is {result2} ')
                if result2 == 'error':
                    assert False
                if result2 == 'complete':
                    if insert_update == compare:
                        assert True
                    else:
                        raise TaskStatisticsError
                    return
            raise TimeoutError

    @staticmethod
    def func_wait_break_point(func1, func2, args, compare, timeout=timeout_config):
        """
        断点续传 stats获取函数
        :param func1:
        :param func2:
        :param args:
        :param compare:
        :param timeout:
        :return:
        """
        time.sleep(check_wait)
        with allure.step('事件统计'):
            for i in range(timeout):
                time.sleep(1)
                result1 = func1(args)
                logger.info(f'{func1.__name__} RESULT is {result1} ')
                result2 = func2(args)
                logger.info(f'task status  is {result2} ')
                if result2 == 'error':
                    assert False
                if result1 >= compare:
                    assert True
            raise TimeoutError


# allure attach
class CTTaskAllureAttach(CTTaskUtilStatusAndStats):
    def generate_task_url(self, task_name, task_type):
        with allure.step('生成任务url'):
            self.generate_task_url_by_name(task_name, task_type)
        if self.task_url:
            allure.attach(
                f'<a href="{self.task_url}" target="_blank">点击此处跳转到task监控界面</a>',
                '任务监控url',
                allure.attachment_type.HTML
            )

    @staticmethod
    def exception_attach(task_name, exception):
        logger.fatal(f'{task_name} {exception}')
        allure.attach(
            f'<h2>{task_name} {exception}</h2>',
            'CASE 错误原因',
            allure.attachment_type.HTML
        )
        raise exception

    def task_log_all(self, task_id):
        log_error = self.get_subtask_info_by_level(task_id, 'ERROR')
        log_warning = self.get_subtask_info_by_level(task_id, 'WARNING')
        log_info = self.get_subtask_info_log_all(task_id)
        allure.attach(
            f'{log_info}',
            'Task LOG ALL',
        )
        if log_error or log_warning:
            allure.attach(
                f'{log_error}\n\n{log_warning}',
                'Task LOG ERROR AND WARNING',
            )


class CTTaskUtil(CTTaskAllureAttach):
    ...


class CTTask(CTTaskUtil):
    ...
